package com.katze.tools.hadoop.config;

import com.katze.tools.hadoop.entity.FileResource;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.URI;
import java.nio.file.Files;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

public class HadoopManager implements Closeable {

    private static final Logger log = LoggerFactory.getLogger(HadoopManager.class);

    private final Configuration configuration;

    private final URI uri;

    private FileSystem hdfs;

    private LocalDateTime timestamp;

    public HadoopManager(Configuration configuration) {
        this.configuration = configuration;
        this.uri = FileSystem.getDefaultUri(configuration);
    }

    protected synchronized void initialization() {
        try {
            if (this.hdfs == null) {
                this.hdfs = FileSystem.get(uri, configuration);
                this.timestamp = LocalDateTime.now();
                Thread thread = new Thread(() -> {
                    try {
                        do {
                            long millis = Duration.between(timestamp, LocalDateTime.now()).toMillis();
                            if (millis > 5 * 60 * 1000) {
                                close();
                                log.info("Hadoop: {} connection closed", uri);
                                break;
                            } else {
                                TimeUnit.SECONDS.sleep(1);
                            }
                        } while (true);
                    } catch (Throwable e) {
                        log.error("HadoopManager-AutoClose-Thread run exception", e);
                        Thread.currentThread().interrupt();
                    }
                });
                thread.setName("HadoopManager-AutoClose-Thread");
                thread.start();
            }
            log.info("Hadoop: {} connection success", uri);
        } catch (IOException e) {
            log.error("Hadoop: {} connection fail", uri, e);
        }
    }

    protected <T>T get(Function<FileSystem, T> call) {
        if (hdfs == null) {
            this.initialization();
        } else {
            this.timestamp = LocalDateTime.now();
        }
        return call.apply(hdfs);
    }

    protected <T>T instance(Function<FileSystem, T> call) {
        try (FileSystem fileSystem = FileSystem.get(uri, configuration)){
            return call.apply(fileSystem);
        } catch (IOException e) {
            log.error("实例化Hadoop: {} 连接对象失败", uri, e);
            return null;
        }
    }

    /**
     * 加载路径信息
     * @param path hadoop 路径（文件/目录路径）
     */
    public FileStatus get(String path, String... name) {
        return get(fs -> {
            try {
                return fs.getFileStatus(name.length ==1? new Path(path, name[0]):new Path(path));
            } catch (IOException e) {
                log.error("加载{}信息失败", path, e);
                throw new RuntimeException("加载" + path + "信息失败");
            }
        });
    }

    /**
     * 文件上传
     * @param file 本地文件
     * @param directory Hadoop路径
     */
    public String upload(String directory, File... file) {
        return get(fs -> {
            for (File item : file) {
                try (FileInputStream in = new FileInputStream(item); FSDataOutputStream out = fs.create(new Path(directory + "/" + item.getName()));) {
                    byte[] buffer = new byte[4096];
                    for (int len = in.read(buffer); len != -1; len = in.read(buffer)) {
                        out.write(buffer, 0, len);
                    }
                } catch (IOException e) {
                    log.error("{}文件上传至{}目录失败", item.getAbsolutePath(), directory, e);
                    if (e instanceof FileAlreadyExistsException) {
                        throw new RuntimeException("当前位置已存在同名文件");
                    }
                    return item.getName();
                }
            }
            return null;
        });
    }

    public void upload(String directory, String name, boolean append, byte[] data) {
        get(fs -> {
            Path path = new Path(directory, name);
            try (FSDataOutputStream out = append? fs.append(path):fs.create(path, true)) {
                out.write(data);
            } catch (IOException e) {
                log.error("{}文件上传至{}目录失败", name, directory, e);
                if (e instanceof FileAlreadyExistsException) {
                    throw new RuntimeException("当前位置已存在同名文件");
                }
                throw new RuntimeException(name + "上传失败");
            }
            return true;
        });
    }

    /**
     * 下载文件
     * @param source Hadoop文件
     * @param directory 本地目录
     */
    public void download(String directory, String... source) {
        get(fs -> {
            for (String item : source) {
                Path file = new Path(item);
                try (FileOutputStream out = new FileOutputStream(new File(directory, file.getName())); FSDataInputStream in = fs.open(file);) {
                    byte[] buffer = new byte[4096];
                    for (int len = in.read(buffer); len != -1; len = in.read(buffer)) {
                        out.write(buffer, 0, len);
                    }
                } catch (IOException e) {
                    log.error("{}文件下载至{}目录失败", source, directory, e);
                    throw new RuntimeException(item + "下载失败");
                }
            }
            return true;
        });
    }

    /**
     * 下载文件
     * @param dir 本地目录
     * @param source Hadoop文件
     */
    public void copyToLocalDir(File dir, String... source) {
        get(fs -> {
            if (dir.exists() || dir.mkdirs()) {
                byte[] buffer = new byte[1024 * 1024 * 50];
                try {
                    for (String item : source) {
                        Path file = new Path(item);
                        FileStatus status = fs.getFileStatus(file);
                        List<FileResource> list;
                        if (status.isDirectory()) {
                            list = listFiles(item, true);
                        } else {
                            list = Collections.singletonList(new FileResource(status));
                        }
                        int last = item.lastIndexOf(Path.SEPARATOR);
                        for (FileResource resource : list) {
                            File parent = new File(dir, resource.getLocation().substring(last));
                            if (!parent.exists()) parent.mkdirs();
                            try (FSDataInputStream in = fs.open(new Path(resource.getPath())); OutputStream out = Files.newOutputStream(parent.toPath().resolve(resource.getName()))){
                                for (int len = in.read(buffer); len != -1; len = in.read(buffer)) {
                                    out.write(buffer, 0, len);
                                }
                            } catch (Exception e) {
                                throw new Exception(resource.getPath());
                            }
                        }
                    }
                } catch (Exception e) {
                    log.error("下载Hadoop文件失败", e);
                    throw new RuntimeException(e.getMessage() + "下载失败");
                }
            }
            return true;
        });
    }

    /**
     * 获取当前路径下所有文件和目录
     * @param directory Hadoop目录
     */
    public List<FileResource> list(String directory) {
        return get(fs -> {
            List<FileResource> list= new ArrayList<>();
            try {
                RemoteIterator<LocatedFileStatus> iterator = fs.listLocatedStatus(new Path(directory));
                while (iterator.hasNext()) {
                    list.add(new FileResource(iterator.next()));
                }
            } catch (IOException e) {
                log.error("查看{}路径失败", directory, e);
            }
            return list;
        });
    }

    public List<FileResource> list(String directory, HdfsFileType type) {
        return get(fs -> {
            List<FileResource> list= new ArrayList<>();
            try {
                RemoteIterator<FileStatus> iterator = fs.listStatusIterator(new Path(directory));
                while (iterator.hasNext()) {
                    FileStatus status = iterator.next();
                    if (type.value(status)){
                        list.add(new FileResource(status));
                    }
                }
            } catch (IOException e) {
                log.error("查看{}路径失败", directory, e);
            }
            return list;
        });
    }


    /**
     * 遍历全部文件，并返回所有文件路径
     * @param directory Hadoop目录
     * @param recursive 是否递归遍历
     */
    public List<FileResource> listFiles(String directory, boolean recursive) {
        return get(fs -> {
            List<FileResource> list= new ArrayList<>();
            try {
                RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path(directory), recursive);
                while (iterator.hasNext()) {
                    list.add(new FileResource(iterator.next()));
                }
            } catch (IOException e) {
                log.error("{}查看失败", directory, e);
            }
            return list;
        });
    }

    /**
     * 创建目录
     * @param directory 路径(包括目录名称)
     */
    public Boolean mkdirs(String directory) {
        return get(fs -> {
            try {
                return fs.mkdirs(new Path(directory));
            } catch (IOException e) {
                log.error("{}创建失败", directory, e);
                return false;
            }
        });
    }

    /**
     * 重命名
     * @param path 文件路径(包括文件名称)
     * @param name 新名称
     */
    public Boolean rename(String path, String name) {
        return get(fs -> {
            try {
                Path src = new Path(path);
                Path dest = new Path(src.getParent(), name);
                if (fs.exists(dest)) {
                    throw new RuntimeException("当前位置已存在同名文件");
                } else {
                    return fs.rename(src, dest);
                }
            } catch (IOException e) {
                log.error("{}重命名失败", path, e);
                throw new RuntimeException("重命名失败");
            }
        });
    }

    /**
     * 移动文件
     * @param directory Hadoop目标目录
     * @param path Hadoop源文件路径
     */
    public Boolean move(String directory, String... path) {
        return get(fs -> {
            boolean rs = true;
            for (String item : path) {
                Path src = new Path(item);
                Path dest = new Path(directory, src.getName());
                try {
                    if (fs.exists(dest)) {
                        throw new RuntimeException("当前位置已存在同名文件");
                    } else {
                        if (!src.getParent().toUri().getPath().equals(directory)) {
                            rs &= fs.rename(src, dest);
                        }
                    }
                } catch (IOException e) {
                    log.error("{}移动失败", path, e);
                    throw new RuntimeException(src.getName() + "移动至" + directory + "失败");
                }
            }
            return rs;
        });
    }

    /**
     * 复制文件
     * @param directory Hadoop目标路径
     * @param path Hadoop源文件路径
     */
    public Boolean copy(String directory, String... path) {
        return get(fs -> {
            boolean rs = true;
            for (String item : path) {
                Path src = new Path(item);
                Path dest = new Path(directory, src.getName());
                try {
                    if (fs.exists(dest)) {
                        throw new RuntimeException("当前位置已存在同名文件");
                    } else {
                        try (FSDataInputStream inputStream = fs.open(src); FSDataOutputStream outputStream = fs.create(dest)){
                            IOUtils.copyBytes(inputStream, outputStream, 4096, false);
                        }
                    }
                } catch (IOException e) {
                    log.error("{}复制失败", path, e);
                    throw new RuntimeException(src.getName() + "复制" + directory + "失败");
                }
            }
            return rs;
        });
    }


    /**
     * 删除文件/目录
     * @param path 路径
     * @param recursive 是否递归遍历
     */
    public Boolean delete(boolean recursive, String... path) {
        return get(fs -> {
            boolean rs = true;
            for (String item : path) {
                Path src = new Path(item);
                try {
                    if (fs.exists(src)) {
                        rs &= fs.delete(src, recursive);
                    }
                } catch (IOException e) {
                    log.error("{}删除失败", path, e);
                    throw new RuntimeException(src.getName() + "删除失败");
                }
            }
            return rs;
        });
    }

    /**
     * 关闭连接
     */
    @Override
    public synchronized void close() throws IOException {
        if (this.hdfs != null) {
            this.hdfs.close();
            this.hdfs = null;
        }
    }
}
